package org.suyafei.func;

import com.alibaba.fastjson.JSONObject;
import org.apache.commons.lang.StringUtils;
import org.apache.flink.api.common.functions.RichFilterFunction;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.configuration.Configuration;
import org.suyafei.util.TimeUtil;

/**
 * @author 宿某人
 * @Date: 2022/04/01/16:22
 * @Description:
 */
public class UniqueVisitorFilterFunction extends RichFilterFunction<JSONObject> {

    private ValueState<String> lastVisitDateState = null;


    @Override
    public void open(Configuration parameters) throws Exception {

        ValueStateDescriptor<String> visitDateState = new ValueStateDescriptor<>("visitDateState", String.class);

//        统计UV
        StateTtlConfig conf = StateTtlConfig.newBuilder(Time.seconds(30)).build();

        visitDateState.enableTimeToLive(conf);

        this.lastVisitDateState = getRuntimeContext().getState(visitDateState);

    }



    @Override
    public void close() throws Exception {
        super.close();
    }

    @Override
    public boolean filter(JSONObject jsonObj) throws Exception {

        Long visiTime = jsonObj.getLong("visitTime");
        String udid = jsonObj.getString("udid");

        //获取当前访问时间
        String currentVisitDate = TimeUtil.format(visiTime);

        String lastVisitDate = lastVisitDateState.value();

        if (StringUtils.isNotEmpty(lastVisitDate) && currentVisitDate.equalsIgnoreCase(lastVisitDate)) {
            System.out.println(udid+"已经在"+lastVisitDate+"访问过");
            return false;
        }else {
            System.out.println(udid+"在"+currentVisitDate+"初次访问");
            lastVisitDateState.update(currentVisitDate);
            return true;
        }
    }
}
